package com.lry.core;

import com.lry.entity.AppendEntriesRequest;
import com.lry.entity.AppendEntriesResponse;
import com.lry.entity.VoteRequest;
import com.lry.entity.VoteResponse;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

//https://hardcore.feishu.cn/docs/doccnMRVFcMWn1zsEYBrbsDf8De
//https://github.com/maemual/raft-zh_cn/blob/master/raft-zh_cn.md
@Slf4j
@Getter
@Setter
public class RaftNode {


    public enum NodeState {
        STATE_FOLLOWER,
        STATE_PRE_CANDIDATE,
        STATE_CANDIDATE,
        STATE_LEADER
    }

    private StateMachine stateMachine;

    private List<Server> servers;

    private List<Peer> peers = new ArrayList<>();
    private Map<Integer,Peer> serverId2Peer;

    private Server localServer;
    private Integer serverId;

    private Integer leaderId;

    private volatile NodeState state = NodeState.STATE_FOLLOWER;

    //服务器当前的任期，逻辑时钟
    private long currentTerm;

    //投票给了谁
    private volatile Integer votedFor;

    private RaftLog raftLog = new MemoryRaftLog();

    //如果commitIndex > lastApplied，则 lastApplied 递增，并将log[lastApplied]应用到状态机中
    //已知已提交的最高的日志条目的索引（初始值为0，单调递增）
    private long commitIndex;

    //已经被应用到状态机的最高的日志条目的索引（初始值为0，单调递增）
    private long lastApplied;

    //用于节点间通信的线程池
    private ExecutorService executorService;

    //定时任务
    private ScheduledExecutorService scheduledExecutorService;

    //选举定时器
    private ScheduledFuture electionScheduledFuture;

    //心跳定时器
    private ScheduledFuture heartbeatScheduledFuture;

    //raft配置
    private RaftOptions raftOptions;


    private Lock lock = new ReentrantLock();
    private Condition commitIndexCondition = lock.newCondition();



    public RaftNode(RaftOptions raftOptions,Server localServer,List<Server> servers,StateMachine stateMachine) {
        assert null!=localServer;
        assert null!=servers && !servers.isEmpty();
        assert null!=raftOptions;
        assert null!=stateMachine;
        this.raftOptions = raftOptions;
        this.stateMachine = stateMachine;
        this.localServer = localServer;
        this.serverId = localServer.getServerId();
        this.servers = servers;
    }

    public void init() {
        serverId2Peer = new ConcurrentHashMap<>();

        for (Server server : this.servers) {
            if(server.equals(localServer)){
                continue;
            }
            Peer peer = new Peer(server);
            serverId2Peer.put(server.getServerId(),peer);
            peers.add(peer);
        }

        // init thread pool
        executorService = new ThreadPoolExecutor(
                raftOptions.getRaftConsensusThreadNum(),
                raftOptions.getRaftConsensusThreadNum(),
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>());

        scheduledExecutorService = Executors.newScheduledThreadPool(2);

        //快照
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                takeSnapshot();
            }
        }, raftOptions.getSnapshotPeriodSeconds(), raftOptions.getSnapshotPeriodSeconds(), TimeUnit.SECONDS);

        // start election
        resetElectionTimer();
    }


    private void takeSnapshot() {

    }


    private void resetHeartbeatTimer() {
        if (heartbeatScheduledFuture != null && !heartbeatScheduledFuture.isDone()) {
            heartbeatScheduledFuture.cancel(true);
        }
        heartbeatScheduledFuture = scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                startNewHeartbeat();
            }
        }, raftOptions.getHeartbeatPeriodMilliseconds(), TimeUnit.MILLISECONDS);
    }

    private void startNewHeartbeat() {
        log.info("server={},发送心跳给所有peer",serverId);
        for (Peer peer : peers) {
            appendEntries(peer);
        }
        resetHeartbeatTimer();
    }



    @SneakyThrows
    public boolean replicate(byte[] command) {
        //非主节点不提供写服务
        if (state != NodeState.STATE_LEADER) {
            log.error("I'm not the leader");
            return false;
        }

        //追加到日志条目中
        Log log1 = new Log(command,currentTerm);
        long newLastLogIndex = raftLog.append(log1);

        //同步给其他节点
        for (Peer peer : peers) {
            executorService.submit(()->{
                appendEntries(peer);
            });
        }


        //是否开启异步复制，是的话，直接返回不等待
        if(raftOptions.isAsyncWrite()){
            return true;
        }

        long startTime = System.currentTimeMillis();
        while(lastApplied<newLastLogIndex){
            if(System.currentTimeMillis()-startTime>=raftOptions.getMaxAwaitTimeout()){
                break;
            }
            commitIndexCondition.await(raftOptions.getMaxAwaitTimeout(),TimeUnit.MILLISECONDS);
        }

        log.warn("lastAppliedIndex={} newLastLogIndex={}", lastApplied, newLastLogIndex);
        if(lastApplied<newLastLogIndex) {
            return false;
        }
        return true;
    }

    private void appendEntries(Peer peer) {

        long prevLogIndex = peer.getNextIndex() -1;
        long prevLogTerm = raftLog.getEntryTerm(prevLogIndex);

        AppendEntriesRequest request = AppendEntriesRequest.builder().
                term(currentTerm).
                leaderId(serverId).
                preLogTerm(prevLogTerm).
                preLogIndex(prevLogIndex).build();

        //打包新发送的entry日志条目
        List<Log> newEntries = packEntries(peer.getNextIndex());
        request.setEntries(newEntries);
        //think more todo
        request.setLeaderCommit(Math.min(commitIndex,prevLogIndex+newEntries.size()));

        AppendEntriesResponse response = peer.getRaftConsensusService().appendEntries(request);


        //如果收到的响应的任期比我大，直接成为follower
        if (response.getTerm() > currentTerm) {

            log.info("server={},Received response from server {} " +
                            "in term {} (this server's term was {})", serverId,
                    peer.getServer().getServerId(),
                    response.getTerm(),
                    currentTerm);

            stepDown(response.getTerm());
        }else{
            if(response.isSuccess()){
                //复制成功，则维护从节点的matchIndex和nextIndex
                peer.setMatchIndex(prevLogIndex + newEntries.size());
                peer.setNextIndex(peer.getMatchIndex() + 1);

                //应用状态机
                advanceCommitIndex();
            }else{
                //失败了，减少nextIndex 重试，直到日志匹配 todo
                peer.decrNextIndex();
            }
        }

    }

    private void advanceCommitIndex() {
        long[] matchIndexes = new long[servers.size()];
        int i=0;
        for (Peer peer : peers) {
            long matchIndex = peer.getMatchIndex();
            matchIndexes[i++] = matchIndex;
        }
        matchIndexes[i] = raftLog.getLastLogIndex();

        Arrays.sort(matchIndexes);
        long newCommitIndex = matchIndexes[servers.size() / 2];
        log.debug("newCommitIndex={}, oldCommitIndex={}", newCommitIndex, commitIndex);


        if (raftLog.getEntryTerm(newCommitIndex) != currentTerm) {
            log.debug("newCommitIndexTerm={}, currentTerm={}",
                    raftLog.getEntryTerm(newCommitIndex), currentTerm);
            return;
        }
        if(commitIndex>=newCommitIndex){
            return;
        }

        commitIndex = newCommitIndex;

        //应用到状态机
        for(long idx = commitIndex+1;idx<=newCommitIndex;idx++){
            Log entry = raftLog.getEntry(idx);
            stateMachine.apply(entry.getCommand());

            lastApplied = idx;
        }

        commitIndexCondition.signalAll();
    }

    private List<Log> packEntries(long nextIndex) {
        //计算出索引最后条目
        long lastIndex = Math.min(raftLog.getLastLogIndex(),nextIndex-1+ raftOptions.getMaxLogEntriesPerRequest());

        List<Log> entries = new ArrayList<>();
        for(long idx = nextIndex;idx<=lastIndex;idx++){
            Log entry = raftLog.getEntry(idx);
            entries.add(entry);
        }
        return entries;
    }

    public void resetElectionTimer() {
        //选举定时器没走完就取消掉，因为收到心跳包需要调用这里，发现没走完就要取消定时器
        if (electionScheduledFuture != null && !electionScheduledFuture.isDone()) {
            log.info("Cancel election timer");
            electionScheduledFuture.cancel(true);
        }
        electionScheduledFuture = scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                startVote();
            }
        }, getElectionTimeoutMs(), TimeUnit.MILLISECONDS);

    }

    private int getElectionTimeoutMs() {
        ThreadLocalRandom random = ThreadLocalRandom.current();
        int randomElectionTimeout = raftOptions.getElectionTimeoutMilliseconds() + random.nextInt(0, raftOptions.getElectionTimeoutMilliseconds());
        log.debug("server={} ,new election time is after {} ms", serverId,randomElectionTimeout);
        return randomElectionTimeout;
    }


    private void startVote() {
        log.info("server={},Running start-vote in term {}",serverId, currentTerm);

        currentTerm++;
        state = NodeState.STATE_CANDIDATE;
        votedFor = serverId;
        leaderId = null;

        Log lastLog = raftLog.getLastLog();

        for (Peer peer : peers) {

            peer.setVoteGranted(null);
            VoteRequest voteRequest = VoteRequest.builder().
                    term(currentTerm).
                    candidateId(serverId).
                    lastLogTerm(lastLog!=null?lastLog.getTerm():0).
                    lastLogIndex(lastLog!=null?lastLog.getIndex():0).
                    build();

            peer.getRaftConsensusServiceAsync().requestVote(voteRequest,new VoteResponseCallBack(peer,voteRequest));
        }

        resetElectionTimer();
    }




    private class VoteResponseCallBack implements RpcCallback<VoteResponse> {

        private VoteRequest voteRequest;
        private Peer peer;

        public VoteResponseCallBack(Peer peer,VoteRequest voteRequest) {
            this.peer = peer;
            this.voteRequest = voteRequest;
        }

        @Override
        public void success(VoteResponse voteResponse) {

            peer.setVoteGranted(voteResponse.isVoteGranted());

            //如果当前任期和请求时候的任期不一致 || 状态不是候选人 放弃成为leader的可能性
            if (currentTerm != voteRequest.getTerm() || state != NodeState.STATE_CANDIDATE) {
                log.info("ignore requestVote RPC result");
                return;
            }
            //如果收到的响应的任期比我大，我要放弃竞选，直接成为follower
            if (voteResponse.getTerm() > currentTerm) {
                log.info("Received RequestVote response from server {} " +
                                "in term {} (this server's term was {})",
                        peer.getServer().getServerId(),
                        voteResponse.getTerm(),
                        currentTerm);
                stepDown(voteResponse.getTerm());
            } else {
                //如果返回成功，则统计票数，过半成为leader
                if(voteResponse.isVoteGranted()){

                    log.info("server {} Got vote from server {} for term {}",serverId,
                            peer.getServer().getServerId(), currentTerm);

                    int voteGrantedNum = 0;
                    if (votedFor.intValue()==serverId) {
                        voteGrantedNum += 1;
                    }

                    for (Peer peer1 : peers) {
                        if (null!=peer1.getVoteGranted()&&peer1.getVoteGranted()) {
                            voteGrantedNum += 1;
                        }
                    }
                    log.info("server={},voteGrantedNum={}", serverId,voteGrantedNum);
                    if (voteGrantedNum > (peers.size()+1) / 2) {
                        becomeLeader();
                    }

                }else{
                    log.info("server={},Vote denied by server {} with term {}, my term is {}",serverId,
                            peer.getServer().getServerId(), voteResponse.getTerm(), currentTerm);
                }
            }

        }

        @Override
        public void fail(Throwable throwable) {
            peer.setVoteGranted(new Boolean(false));
        }
    }

    private void becomeLeader() {
        if(state==NodeState.STATE_LEADER){
            return;
        }
        //当一个领导人刚获得权力的时候，他初始化所有的 nextIndex 值为自己的最后一条日志的 index 加 1
        for (Peer peer : peers) {
            peer.setNextIndex(raftLog.getLastLogIndex()+1);
        }
        log.info("Got majority vote, serverId={} become leader", serverId);

        state = NodeState.STATE_LEADER;
        leaderId = serverId;
        // stop vote timer
        if (electionScheduledFuture != null && !electionScheduledFuture.isDone()) {
            electionScheduledFuture.cancel(true);
        }
        // start heartbeat timer
        startNewHeartbeat();
    }


    public void stepDown(long newTerm) {
        log.warn("step down,serverId={},curTerm={},newTerm={}",serverId,currentTerm,newTerm);
        if (currentTerm > newTerm) {
            log.error("server={},currentTerm={},newTerm={},can't be happened",serverId,currentTerm,newTerm);
            return;
        }
        if (currentTerm < newTerm) {
            currentTerm = newTerm;
            leaderId = 0;
            votedFor = 0;
        }
        state = NodeState.STATE_FOLLOWER;
        // stop heartbeat
        if (heartbeatScheduledFuture != null && !heartbeatScheduledFuture.isDone()) {
            heartbeatScheduledFuture.cancel(true);
        }
        resetElectionTimer();
    }
}
